/*=============================================================================
#     FileName: srv_main.cpp
#         Desc: 主函数
#       Author: Ace
#        Email: goace@acm.org
#     HomePage: http://www.goace.co.cc
#      Version: 0.0.1
#   LastChange: 2011-04-10 13:02:20
#      History:
=============================================================================*/
#pragma pack(4)

#include  <sys/epoll.h>
#include  <stdlib.h>
#include  <unistd.h>
#include  <fcntl.h>
#include  <errno.h>
#include  <sys/stat.h>
#include  <sys/types.h>
#include  <sys/wait.h>
#include  <stdio.h>
#include  <string.h>
#include  <iostream>
#include  <semaphore.h>
#include  <sys/sendfile.h>
#include  "db.h"
#include  "my_tcp_lib.h"
#include  "gserver.h"
#include  "event_handler.h"
using namespace std;

//全局变量
DB			db;						//数据库对象
sem_t		ctl_have_events;		//信号量，同步ctl进程中的线程读取事件
int			n_ctl_idle_threads;		//记录空闲的ctl线程


/* ---------------------------------------------------------------*/
/**
 * 函数简介:  把传入的fd设成非阻塞
 *
 * 参数: sock 需要设置的fd
 */
/* ---------------------------------------------------------------*/
void setnonblocking(int sock)
{
    int opts;
    opts=fcntl(sock,F_GETFL);
    if(opts<0)
    {
        perror("fcntl(sock,GETFL)");
        exit(1);
    }
    opts = opts|O_NONBLOCK;
    if(fcntl(sock,F_SETFL,opts)<0)
    {
        perror("fcntl(sock,SETFL,opts)");
        exit(1);
    }   
}

/* ---------------------------------------------------------------*/
/**
 * 函数简介:  建立工作进程，master永不返回，
 *            始终监视子进程的情况，如果子
 *            进程意外crash，就重新建立。
 *
 * 参数: nWorker 要建立的进程数
 *
 * 返回值:  -1表示错误   0表示成功
 *          master进程不会返回
 */
/* ---------------------------------------------------------------*/
int create_workers(int nWorker)
{
    if (0 < nWorker)
    {
        bool bIsChild;
        pid_t nPid;

        while (!bIsChild)
        {
            if (0 < nWorker)
            {
                nPid = ::fork();
                if (nPid > 0)
                {
                    bIsChild = false;
                    --nWorker;
                }
                else if (0 == nPid)
                {
                    bIsChild = true;
                    printf("create worker %d success!\n", getpid());
                }
                else
                {
                    printf("fork error: %s\n", strerror(errno));
                    return -1;
                }
            }
            else 
            {
                int nStatus;
                if (-1 == wait(&nStatus))
                {
                    ++nWorker;
                }
            }
        }
    }

    return 0;
}

/* ---------------------------------------------------------------*/
/**
 * 函数简介:  初始化进程池
 *
 * 参数： n_child 要创建的workers数
 */
/* ---------------------------------------------------------------*/
void data_workers(int n_child)
{
	void data_process(int);

	int listen_sock;
	listen_sock = tcp_listen(NULL,SRV_DATA_PORT,NULL);	
	if( listen_sock < 0 )
	{
		perror("tcp_listen");
		master_exit(-1);
	}

	setnonblocking(listen_sock);
	
	//创建工作进程
	create_workers(n_child);

	data_process(listen_sock);
}

/* ---------------------------------------------------------------*/
/**
 * 函数简介: 数据处理进程的主例程 
 *
 * 三种请求类型的状态机如下：
 *   SYNC： SYNC_WAIT_ID  ->  SYNC_WAIT_VERSION  -> SYNC_SENDING
 *   PUSH:  PUSH_WAIT_ID  ->  PUSH_WAIT_ACK  ->  PUSH_SENDING
 *              ->  PUSH_RECIVING
 *   PULL:  PULL_WAIT_ID  ->  PULL_RECIVING  ->  PULL_WAIT_ACK
 *              ->  PULL_SENDING
 *
 * 参数: listen_sock 监听的socket
 */
/* ---------------------------------------------------------------*/
void data_process(int listen_sock)
{
	int8_t		reply;					//给用户的反馈
	char		name[USR_LEN], pwd[PWD_LEN];
	uint8_t		request_type;

	struct epoll_event ev,events[EPOLL_MAX_EVENTS];
	data_sock_info_t *ptr;

	int nfds,							//计数epoll返回的事件数目
		connfd,							//accept返回的fd
		n_read;							//read返回的数	
	int epfd = epoll_create(EPOLL_MAX_FD);
	if( epfd < 0 )
	{
		data_exit(-1);
	}

	//把监听socket加入epoll集合中
	ev.events = EPOLLIN | EPOLLET;
	ev.data.ptr = new ctl_sock_info_t;
	((data_sock_info_t*)ev.data.ptr)->sock_fd = listen_sock;

	if( epoll_ctl(epfd,EPOLL_CTL_ADD,listen_sock,&ev) < 0 )
	{
		perror("epoll_ctl");
		data_exit(-1);
	}

	//main loop
	while( true )
	{
		nfds = epoll_wait(epfd,events,EPOLL_MAX_EVENTS,-1);
		if( nfds == -1 )
		{
			if( errno == EINTR )
				continue;
			else
			{
				perror("epoll_wait");
				ctl_exit(-1);
			}
		}

		//循环处理每个返回的event
		for( int i = 0 ; i < nfds ; i++ )
		{
			ptr = (data_sock_info_t*)events[i].data.ptr;

			if( ptr->sock_fd == listen_sock )
			{
				//因为设置的ET模式，因此要循环accept直到EAGAIN
				while( true )
				{
					connfd = accept(listen_sock,NULL,NULL);
					if( connfd == -1 )
					{
						if( errno == EINTR )
							continue;
						else if( errno == EAGAIN || errno == EWOULDBLOCK)
							break;
						else if( errno == ECONNABORTED)
							break;
						else
						{
							perror("accept");
							continue;
						}
					}
					
					//把返回的socket加入epoll集合
					setnonblocking(connfd);
					ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;

					//分配结构体
					ev.data.ptr = new data_sock_info_t;
					bzero(ev.data.ptr,sizeof(data_sock_info_t));
					((data_sock_info_t*)ev.data.ptr)->sock_fd = connfd;
					((data_sock_info_t*)ev.data.ptr)->type = DATA_INIT;
					
					if( epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev) < 0 )
					{
						perror("epoll_ctl");
						continue;
					}
				}
			}
			else	//如果是非监听socket的事件
			{
				if( events[i].events & EPOLLIN )	//如果是可读
				{
					if( ptr->type == DATA_INIT )	//如果处在INIT状态
					{
						//读取请求类型：SYNC PULL PUSH之一
						if( read(ptr->sock_fd,&ptr->type,sizeof(ptr->type)) < 0 )
						{
							data_error("read from peer",ptr);
							continue;
						}

						//因为epoll设置了ET模式  所以要继续读  直到EAGAIN
						if( ptr->type == DATA_SYNC )
						{
							cout << "0" << endl;

							//读取用户ID
							if( data_read_id(ptr) < 0 )
								continue;

							cout << "1" << endl;
							//读取客户端版本号
							if( data_read_version(SYNC_WAIT_PEER,ptr) < 0 )
								continue;

							cout << "peer" << (int)ptr->peer_version << endl;
							ptr->peer_version++;

							//读取服务器版本号
							if( data_read_version(SYNC_WAIT_REAL,ptr) < 0 )
								continue;
							cout << "real" << (int)ptr->real_version << endl;

							ptr->state = SYNC_WAIT_REAL;
							
							//打开日志文件
							if( open_log_file(ptr) < 0 )
								continue;

							ptr->state = SYNC_SENDING;
							ptr->transfering = true;
							ptr->tsfd = 0;

							//等待该socket可写
							ev.events = EPOLLOUT | EPOLLRDHUP | EPOLLET;
							ev.data.ptr = ptr;
							if( epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->sock_fd,&ev) < 0 )
							{
								data_error("epoll_ctl",ptr);
								continue;
							}

							cout << "epoll_ctl ok" << endl;
						}
						else if( ptr->type == DATA_PUSH )
						{
							//读取用户ID
							if( data_read_id(ptr) < 0 )
								continue;



						}
						else if( ptr->type == DATA_PULL )
						{

						}

					}
					else if( ptr->type == DATA_SYNC )
					{
						 if( ptr->state == SYNC_WAIT_ID )
						 {
					
							 cout << "返回了"<< endl;
							 //读取用户ID
							 if( data_read_id(ptr) < 0 )
								 continue;

							 cout << "1" << endl;
							 ptr->state == SYNC_WAIT_PEER;
						 }
						 
						 if( ptr->state == SYNC_WAIT_PEER )
						 {
							 //读取客户端版本号
							 if( data_read_version(SYNC_WAIT_PEER,ptr) < 0 )
								 continue;

							 cout << "2" << endl;
							 ptr->state = SYNC_WAIT_REAL; 
						 }

						 if( ptr->state == SYNC_WAIT_REAL )
						 {
							 //读取服务器版本号
							 if( data_read_version(SYNC_WAIT_REAL,ptr) < 0 )
								 continue;
							 cout << "3" << endl;
						 }

						 //打开日志文件
						 if( open_log_file(ptr) < 0 )
							 continue;

						 ptr->state = SYNC_SENDING;
						 ptr->transfering = true;

						 //等待该socket可写
						 ev.events = EPOLLOUT | EPOLLRDHUP | EPOLLET;
						 ev.data.ptr = ptr;
						 if( epoll_ctl(epfd,EPOLL_CTL_MOD,
									 ptr->sock_fd,&ev) < 0 )
						 {
							 data_error("epoll_ctl",ptr);
							 continue;
						 }

					}
					else if( ptr->type == DATA_PUSH )
					{

					}
					else if( ptr->type == DATA_PULL )
					{

					}
				}
				else if( events[i].events & EPOLLOUT )
				{
					cout << "epollout" << endl;
					if( ptr->type == DATA_SYNC)
					{
						cout << "DATA_SYNC" << endl;
						size_t sent;

						while( ptr->tsfd < ptr->total )
						{
							if(( sent = sendfile(ptr->sock_fd,ptr->file_fd,
										NULL, BLOCK_SIZE)) < 0 )
							{
								if( errno == EAGAIN )
									break;

								data_error("sendfile",ptr);
								break;
							}
							cout << "sent " << sent << endl; 
							ptr->tsfd += sent;

							if( ptr->tsfd == ptr->total )
							{
								//如果所有log传送完了
								if( ptr->peer_version - 1 == ptr->real_version )
								{
									cout << "log传送完毕" << endl;
									close(ptr->file_fd);
									close(ptr->sock_fd);
									delete ptr;
									break;
								}
									
								//打开下一个log文件
								close(ptr->file_fd);
								if( open_log_file(ptr) < 0 )
								{
									data_error("open log file",ptr);
									break;
								}
							}
						}
						
					}
					else if( ptr->type == DATA_PUSH )
					{
						
					}
					else if( ptr->type == DATA_PULL )
					{
						
					}
				}
				else
				{

				}
			}
		}
	}
}

/* ---------------------------------------------------------------*/
/**
 * 函数简介:  循环接收并处理用户“控制信息”，包括注册、登录、
 *			  删除、新建
 *
 * 参数: listen_sock 监听的socket
 */
/* ---------------------------------------------------------------*/
void ctl_process(int listen_sock)
{
	int8_t		reply;					//给用户的反馈
	char		name[USR_LEN], pwd[PWD_LEN];
	
	struct epoll_event ev,events[EPOLL_MAX_EVENTS];
	ctl_sock_info_t *ptr;

	int nfds,							//计数epoll返回的事件数目
		connfd,							//accept返回的fd
		n_read;							//read返回的数	

	int epfd = epoll_create(EPOLL_MAX_FD);
	if( epfd < 0 )
	{
		perror("epoll_create");
		ctl_exit(-1);
	}

	//把监听socket加入epoll集合中
	ev.events = EPOLLIN | EPOLLET;
	ev.data.ptr = new ctl_sock_info_t;
	((ctl_sock_info_t*)ev.data.ptr)->fd = listen_sock;

	if( epoll_ctl(epfd,EPOLL_CTL_ADD,listen_sock,&ev) < 0 )
	{
		perror("epoll_ctl");
		ctl_exit(-1);
	}

	//主循环
	while( true )
	{
		nfds = epoll_wait(epfd,events,EPOLL_MAX_EVENTS,-1);
		if( nfds == -1 )
		{
			if( errno == EINTR )
				continue;
			else
			{
				perror("epoll_wait");
				ctl_exit(-1);
			}
		}

		//循环处理每个返回的event
		for( int i = 0 ; i < nfds ; i++ )
		{
			ptr = (ctl_sock_info_t*)events[i].data.ptr;

			//如果返回的是监听socket
			if( ptr->fd == listen_sock )
			{
				//因为设置的ET模式，因此要循环accept直到EAGAIN
				while( true )
				{
					connfd = accept(listen_sock,NULL,NULL);
					if( connfd == -1 )
					{
						if( errno == EINTR )
							continue;
						else if( errno == EAGAIN || errno == EWOULDBLOCK)
							break;
						else if( errno == ECONNABORTED)
							break;
						else
						{
							perror("accept");
							continue;
						}
					}
					
					//把返回的socket加入epoll集合
					setnonblocking(connfd);
					ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;

					//分配结构体
					ev.data.ptr = new ctl_sock_info_t;
					bzero(ev.data.ptr,sizeof(ctl_sock_info_t));
					((ctl_sock_info_t*)ev.data.ptr)->fd = connfd;
					
					if( epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev) < 0 )
					{
						perror("epoll_ctl");
						continue;
					}
				}
			} 
			else
			{
				if( events[i].events & EPOLLIN )
				{
					ctl_pkg_t *pkg_in = new ctl_pkg_t;

					if ( (n_read = readn(ptr->fd,pkg_in,sizeof(*pkg_in))) <= 0)
					{
						if( n_read == 0 )
						{
							//返回0说明对方关闭了连接
							usr_exit(ptr);
						}
						else
						{
							perror("ctl read from client");
							usr_exit(ptr);
						}
					}

					//处理事件
					if( pkg_in->type == REG )
					{
						strncpy(ptr->usr_name,pkg_in->str.id.usr,USR_LEN);
						ptr->reply = db.add_usr(ptr->usr_name,pkg_in->str.id.pwd);
						
#ifdef DEBUG
						cout << ptr->usr_name << " regist pwd:" 
							 << pkg_in->str.id.pwd << endl;
#endif

						if(ptr->reply == 0)
						{
							//如果注册成功
							//建立用户主目录：/gsync/user_ID
							string main_path = SRV_ROOT;
							main_path += "/";
							main_path += pkg_in->str.id.usr;

							if( mkdir(main_path.c_str(),0766) < 0 )
							{
								perror("make usr main_path");
								close(ptr->fd);
								delete(ptr);
								continue;
							}

							//创建.log目录，存放log文件
							string log_dir = main_path + "/.log";
							if( mkdir(log_dir.c_str(),0766) < 0 )
							{
								perror("make .log dir");
								close(ptr->fd);
								delete ptr;
								continue;
							}

							//创建version文件
							//version记录当前版本号，初始化为0
							string version_file = main_path + "/.version";
							int vf_fd = open(version_file.c_str(),O_WRONLY|O_CREAT,0766);
							if( vf_fd < 0 )
								perror("create version_file");

							int32_t init_version = 0;
							if( write(vf_fd,&init_version,sizeof(int32_t)) < 0 )
								perror("write to version file");

							close(vf_fd);
						}
						
						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = REG;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("write to user");
						}

						close(ptr->fd);
						delete(ptr);
						delete(pkg_in);
					}
					else if( pkg_in->type == LOGIN ) // 用户登录 
					{
						strncpy(ptr->usr_name,pkg_in->str.id.usr,USR_LEN);
						ptr->reply = db.log_in(ptr->usr_name,pkg_in->str.id.pwd);
						
						if( ptr->reply == 0)
						{
							//生成版本号文件的路径
							char version_file[USR_LEN];
							snprintf(version_file,USR_LEN,"%s/%s/.version",
									SRV_ROOT,ptr->usr_name);
							
							//打开并读取版本号
							ptr->updated = false;
							int vf_fd = open(version_file,O_RDONLY);
							if( vf_fd < 0 )
							{
								perror("open version file");
								close(ptr->fd);
								delete(ptr);
								continue;
							}

							reply = read(vf_fd,&(ptr->version),sizeof(int32_t));
							if( reply < 0 )
							{
								perror("read from version file");
								close(ptr->fd);
								delete(ptr);
								continue;
							}

							close(vf_fd);

#ifdef DEBUG
							cout << ptr->usr_name << " login at version " 
								 << ptr->version << endl;
#endif

						}

						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = LOGIN;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("write to user");
							close(ptr->fd);
							delete(ptr);
							continue;
						}

						//发送版本号
						write(ptr->fd,&(ptr->version),sizeof(ptr->version));

						//如果登录失败
						if( ptr->reply < 0 )
						{
							close(ptr->fd);
							delete(ptr);
							continue;
						} 

						delete(pkg_in);
					}
					else if( pkg_in->type == CREATE_D ) //用户创建目录
					{
						//防止空串
						if( strlen(pkg_in->str.path) == 0 )
							continue;

						//创建目录
						ptr->reply = mkdir(pkg_in->str.path,0766);
						if( ptr->reply < 0 )
							perror("make new dir");

#ifdef DEBUG
						cout << ptr->usr_name << " creat dir "
							 <<	pkg_in->str.path << endl;
#endif
						
						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = CREATE_D;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("write to user");
						}
						
						ptr->history.push_back(pkg_in);
					}
					else if( pkg_in->type == CREATE_F ) //用户创建文件
					{
						//防止空串
						if( strlen(pkg_in->str.path) == 0 )
							continue;

						//创建目录
						int new_file = open(pkg_in->str.path,O_CREAT|O_RDWR,0766);
						if( new_file < 0 )
						{
							perror("touch new file");
							ptr->reply = -1;
						}
						else
							ptr->reply = 0;

						close(new_file);
						
#ifdef DEBUG
						cout << ptr->usr_name << " creat file "
							 <<	pkg_in->str.path << endl;
#endif
						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = CREATE_F;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("write to user");
						}
						
						ptr->history.push_back(pkg_in);
					}
					else if( pkg_in->type == DEL_F ) //用户删除文件
					{
						//防止空串
						if( strlen(pkg_in->str.path) == 0 )
							continue;

						//删除文件
						ptr->reply = unlink(pkg_in->str.path);
						if( ptr->reply < 0 )
						{
							perror("remove file");
						}

#ifdef DEBUG
						cout << ptr->usr_name << " remove file "
							 <<	pkg_in->str.path << endl;
#endif
						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = DEL_F;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("write to user");
						}

						ptr->history.push_back(pkg_in);
					}
					else if( pkg_in->type == DEL_D ) // 用户删除目录
					{
						//防止空串
						if( strlen(pkg_in->str.path) == 0 )
							continue;

						//删除文件
						ptr->reply = rmdir(pkg_in->str.path);
						if( ptr->reply < 0 )
						{
							perror("remove dir");
						}

#ifdef DEBUG
						cout << ptr->usr_name << " remove dir "
							 <<	pkg_in->str.path << endl;
#endif
						
						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = DEL_D;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("write to user");
						}

						ptr->history.push_back(pkg_in);
					}
					else if( pkg_in->type == UP_F )
					{
						;
					}
					else if( pkg_in->type == UP_D )
					{

					}
					else if( pkg_in->type ==  EXIT)
					{
						usr_exit(ptr);
					}


				} 
				else if( events[i].events & EPOLLOUT)
				{
					if( write(ptr->fd,&ptr->reply,sizeof(int8_t)) < 0 )
					{
						perror("ctl write to client");
						usr_exit(ptr);
					}

					if( ptr->state == REG )
					{
						close(ptr->fd);
						delete(ptr);
					}
					else if(ptr->state == LOGIN && ptr->reply < 0)
					{
						close(ptr->fd);
						delete(ptr);
					}
					else
					{
						events[i].events = EPOLLIN | EPOLLET | EPOLLRDHUP;
						if( epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]) < 0 )
						{
							perror("epoll_ctl");
							usr_exit(ptr);
						}
					}
				}
				else if( events[i].events & EPOLLRDHUP )
				{
					usr_exit(ptr);
				}
			}
		}
		
	}

}

/* ---------------------------------------------------------------*/
/**
 * 函数简介:  接受控制信息的进程，控制信息包括新建、删除
 *            更新等
 *
 * 参数：n_child 要创建的worker数
 */
/* ---------------------------------------------------------------*/
void ctl_workers(int n_child)
{
	int listen_sock;
	listen_sock = tcp_listen(NULL,SRV_CTL_PORT,NULL);	
	if( listen_sock < 0 )
	{
		perror("tcp_listen");
		ctl_exit(-1);
	}

	setnonblocking(listen_sock);
	
	//创建多个进程拷贝
	create_workers(n_child);

	//创建线程池
	//pool_init();

	ctl_process(listen_sock);
}



int main(int argc, char *argv[])
{
	//连接数据库
	if( db.init() < 0 )
	{
		exit(-1);
	}
	
	//分裂成两个进程
	pid_t pid = fork();
	if( pid < 0 )
	{
		perror("fork error");
		exit(-1);
	}
	else if( pid == 0 )
	{
		//进程池管理员
		data_workers(2);
	}
	else
	{
		//控制进程
		ctl_workers(2);
	}
}

